package com.example.flinkcourse.lesson5;

import com.example.flinkcourse.lesson5.config.Config;
import com.example.flinkcourse.lesson5.evolution.SchemaEvolution;
import com.example.flinkcourse.lesson5.route.DataRouter;
import com.example.flinkcourse.lesson5.sink.DorisSink;
import com.example.flinkcourse.lesson5.source.MySqlCdcSource;
import com.example.flinkcourse.lesson5.transform.DataTransform;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Flink CDC 实战教程主应用类
 */
public class Lesson5Application {
    private static final Logger LOG = LoggerFactory.getLogger(Lesson5Application.class);

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 加载配置
        Config config = Config.createDefault();
        
        // 配置执行环境
        env.setParallelism(config.getFlinkParallelism());
        env.enableCheckpointing(config.getFlinkCheckpointInterval());
        
        LOG.info("Flink CDC 实战教程启动...");
        
        // 根据命令行参数选择要运行的示例
        if (args.length > 0) {
            switch (args[0]) {
                case "mysql-to-doris":
                    runMySqlToDorisExample(env, config);
                    break;
                case "mysql-to-starrocks":
                    runMySqlToStarRocksExample(env, config);
                    break;
                default:
                    LOG.error("未知的示例类型: {}", args[0]);
                    System.exit(1);
            }
        } else {
            LOG.error("请指定要运行的示例类型");
            System.exit(1);
        }
    }
    
    /**
     * 运行 MySQL 到 Doris 同步示例
     */
    private static void runMySqlToDorisExample(StreamExecutionEnvironment env, Config config) throws Exception {
        LOG.info("开始运行 MySQL 到 Doris 同步示例...");
        
        // 创建 MySQL CDC Source
        MySqlCdcSource.createSource(env, config)
            // 添加 Schema Evolution 处理
            .map(SchemaEvolution.createEvolution())
            .name("schema-evolution")
            // 添加数据路由
            .flatMap(DataRouter.createRouter())
            .name("data-router")
            // 添加数据转换
            .map(DataTransform.createTransform())
            .name("data-transform")
            // 添加 Doris Sink
            .addSink(DorisSink.createSink(config, "orders"))
            .name("orders-sink");
            
        // 执行作业
        env.execute("MySQL to Doris Sync");
    }
    
    /**
     * 运行 MySQL 到 StarRocks 同步示例
     */
    private static void runMySqlToStarRocksExample(StreamExecutionEnvironment env, Config config) throws Exception {
        LOG.info("开始运行 MySQL 到 StarRocks 同步示例...");
        // TODO: 实现 MySQL 到 StarRocks 同步逻辑
    }
} 